package com.github.binarylei.hadoop.mr.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;

/**
 * 用来描述一个特定的作业
 * 比如，该作业使用哪个类作为逻辑处理中的 map，哪个作为 reduce
 * 还可以指定该作业要处理的数据所在的路径
 * 还可以指定改作业输出的结果放到哪个路径
 */
public class WCRunner {

    public static void main(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir", "D:/SoftWare/BigData/Hadoop/winutils/hadoop-2.8.3");
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "file:///E:/test/");
        Path input = new Path("file:///E:/test/input");
        Path output = new Path("file:///E:/test/output");
        FileUtil.fullyDelete(new File("E:/test/output"));
        FileUtil.fullyDelete(new File("E:/tmp"));

        Job wcjob = Job.getInstance(conf);
        // 设置整个 job 所用的那些类在哪个 jar 包
        wcjob.setJarByClass(WCRunner.class);
        wcjob.setJobName("WCRunner");
//
//		FileSystem fs = wcjob.getCluster().getFileSystem();
//		fs.deleteOnExit(output);

        // 本 job 使用的 mapper 和 reducer 的类
        wcjob.setMapperClass(WCMapper.class);
        wcjob.setReducerClass(WCReducer.class);

        // 指定 mapper 的输出数据 kv 类型
        wcjob.setInputFormatClass(TextInputFormat.class);
        wcjob.setMapOutputKeyClass(Text.class);
        wcjob.setMapOutputValueClass(LongWritable.class);

        // 指定 reduce 的输出数据 kv 类型
        wcjob.setNumReduceTasks(1);
        wcjob.setOutputKeyClass(Text.class);
        wcjob.setOutputValueClass(LongWritable.class);

        // 指定要处理的输入数据存放路径
        FileInputFormat.setInputPaths(wcjob, input);
        // 指定处理结果的输出数据存放路径
        FileOutputFormat.setOutputPath(wcjob, output);

        // 将 job 提交给集群运行
        wcjob.waitForCompletion(true);
    }
}
